package cn.krui.project.pubsub.system;

import redis.clients.jedis.JedisPubSub;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.util.StringUtils;

import cn.krui.helper.tools.JsonUtils;
import cn.krui.helper.tools.RedisUtils;
import cn.krui.helper.tools.WebToolsUtils;

/**
 * @author zc
 * system 模块消息处理
 */
public class SubscriberSystem extends JedisPubSub {
    private static final Logger logger = LoggerFactory.getLogger(SubscriberSystem.class);

    private StringRedisTemplate redisTemplate;

    private final String EQUALS_CHECK = "check";

    public SubscriberSystem(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @Override
    public void onMessage(String channel, String message) { // 收到消息会调用
        logger.info(String.format("收到redis发布的消息,频道 %s, 内容 %s", channel, message));
        if (StringUtils.isEmpty(message)){
            return;
        }
        ObjectMapper mapper = new ObjectMapper();
        try {
            JsonNode node = mapper.readTree(message);
            String msgType = node.path("type").asText();
            String data = node.path("data").asText();
            String msgUuid = node.path("msg_uuid").asText();
            logger.info(msgType);
            if(!data.equals(EQUALS_CHECK)){
                // zc 0307 有消费端保存消息到队列
                // 1：增加队列保存后缀，区分不同的消费端
                // 2：通过消息的uuid在不同的消费端验证队列去重
                // *注意_0是这类消息的一种消费方式，用于处理单消息多订阅处理，队列做了同模块消息去重
                RedisUtils redisUtils = new RedisUtils(redisTemplate);
                if(!redisUtils.setIfAbsentEx(
                    String.format("pub:%s:$s_list_0:%s", channel, msgType, msgUuid),"", 10, TimeUnit.SECONDS)){
                    //10秒key用于去重，此处不要debug
                    return; 
                }
                redisUtils.lRightPush(String.format("pub:%s:%s_list_0", channel, msgType), JsonUtils.toJson(node.get("data")));
            }
            Map<String, String> header = new HashMap<>(1);
            header.put("super", "mq");
            String outStr = WebToolsUtils.httpPost(node.path("exec_url").asText(), new HashMap<>(0), header);
            logger.info(String.format("%s通知成功：%s", msgType, outStr));
        } catch (IOException e) {
            logger.error("消息处理：system %s", e);
        }
    }

    @Override
    public void onSubscribe(String channel, int subscribedChannels) { // 订阅了频道会调用
        logger.info(String.format("成功订阅redis频道,频道 %s,订阅频道数%d", channel, subscribedChannels));
    }

    @Override
    public void onUnsubscribe(String channel, int subscribedChannels) { // 取消订阅 会调用
        logger.info(String.format("取消redis频道订阅,频道 %s,订阅频道数 %d", channel, subscribedChannels));

    }
}